package streamforce.crypto.owner;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Properties;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import streamforce.Stat;
import streamforce.crypto.KPAbe;
import streamforce.crypto.event.StockEvent;
import streamforce.crypto.message.CloseConnection;
import streamforce.crypto.message.PolicyMessage;
import streamforce.workload.PolicyInfo;

//Connect to the cloud server 
public class EPOwnerClient {
	private static final Log log = LogFactory.getLog(EPOwnerClient.class);
	private static final String ENC_FILE="enc_file";
	String serverHost="";
	int serverPort=0, ownerId=0;	
	ObjectOutputStream oos = null;
	ObjectInputStream ois = null;
	KPAbe cryptoHandler=null; 
		
	int streamId; 
	byte[] pK, detK; 
	int[] windowSizes = null;
	DataEncryptor encryptor; 
	int aggregateType;
	
	Stat stat; 		
	
	public EPOwnerClient(int streamId){
		this.streamId = streamId; 
		stat = new Stat("ownerThroughput");
		
		Properties prop = new Properties();
		try {
			prop.load(new FileInputStream("attributes.properties"));
			this.serverHost = prop.getProperty("serverHost");
			this.serverPort = new Integer(prop.getProperty("serverPort"))
					.intValue();
			Socket socket = new Socket(serverHost, serverPort);
			this.oos = new ObjectOutputStream(
					socket.getOutputStream());		
			this.ois = new ObjectInputStream(socket.getInputStream()); 
			this.cryptoHandler = new KPAbe(); 
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			//e.printStackTrace();
			System.err.println("Connection to the server failed");
		}
		long start = System.currentTimeMillis(); 
		this.initialization(); 
		long end = System.currentTimeMillis();
		System.out.println("owner initialization time = "+(end-start));
		encryptor = new DataEncryptor(this.streamId, this.cryptoHandler,
				this.windowSizes);
	}
	
	public void waitForAck(){		
		try {
			this.ois.readObject();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} 
	}
	
	public void sendPolicy(PolicyMessage msg){
		try {
			this.oos.writeObject(msg);
			this.oos.flush();
						
		} catch (IOException e) {
			System.err.println("Fail writing Policy Message"); 
		} 
	}
	
	public void closeConnection(){
		try{
			this.oos.writeObject(new CloseConnection());			
		}
		catch (Exception e){
			e.printStackTrace(); 
		}
	}
	public StockEvent encryptData(int timestamp, int hour, int stockId, int price, int volume){
		StockEvent sd=null; 		
		switch (this.aggregateType){
		case -1: 
			sd = encryptor.encrypt_simple(timestamp, hour, stockId, price, volume);			
			break;
		case 1:
			sd = encryptor.encrypt_Aggregate1(timestamp, hour, stockId, price, volume);			
			break;
		case 2:
			sd = encryptor.encrypt_Aggregate2(timestamp, hour, stockId, price, volume);			
			break;
		case 3:
			sd = encryptor.encrypt_Aggregate3(timestamp, hour, stockId, price, volume);			
			break;
		}				
		encryptor.encrypt_join_price(sd, price);
		encryptor.encrypt_join_volume(sd, volume);	
		this.stat.update();
		return sd; 
	}
	
	public void encryptAndSendData(int timestamp, int hour, int stockId, int price, int volume){
		StockEvent sd=null; 		
		switch (this.aggregateType){
		case -1: 
			sd = encryptor.encrypt_simple(timestamp, hour, stockId, price, volume);			
			break;
		case 1:
			sd = encryptor.encrypt_Aggregate1(timestamp, hour, stockId, price, volume);			
			break;
		case 2:
			sd = encryptor.encrypt_Aggregate2(timestamp, hour, stockId, price, volume);			
			break;
		case 3:
			sd = encryptor.encrypt_Aggregate3(timestamp, hour, stockId, price, volume);			
			break;
		}				
		encryptor.encrypt_join_price(sd, price);
		encryptor.encrypt_join_volume(sd, volume);
		this.stat.update();
		
		try {
			this.oos.writeObject(sd);
			this.oos.flush();
		} catch (IOException e) {
			System.err.println("Fail writing objects"); 
			e.printStackTrace();
		} 		 
	}
	
	private void initialization(){
		Properties prop = new Properties();
		try {
			prop.load(new FileInputStream("attributes.properties"));			
			int nAttributes = new Integer(prop.getProperty("nStocks")).intValue();
			String[] ws = prop.getProperty("windowSizes").split(",");
			String[] universe = new String[nAttributes+6+(ws.length)];
			int i=0;
			for (i=0; i<nAttributes; i++)
				universe[i] = "stockId_"+i;			
			universe[i] = "timestamp = ";i++;
			universe[i] = "hour = "; i++;
			universe[i] = "ws_x";i++;//all window sizes > 1
			universe[i] = "ws_1";i++;
			universe[i] = "att_price";i++;
			universe[i] = "att_volume";i++;
						
			windowSizes = new int[ws.length];
			for (int j=0; j<ws.length; j++){
				windowSizes[j] = new Integer(ws[j]).intValue();
				universe[i+j] = "ws_"+windowSizes[j];
			}						
			
			cryptoHandler = new KPAbe();
			this.pK = cryptoHandler.initOwner(this.streamId, universe, 2, windowSizes);
			this.detK = cryptoHandler.generateUserDetKey(this.streamId);
			if (prop.containsKey("aggregateType") && !prop.getProperty("aggregateType").isEmpty())
				this.aggregateType = new Integer(prop.getProperty("aggregateType")).intValue();
			else
				this.aggregateType = -1;
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public static void sendData(EPOwnerClient owner, int streamId){
		int ts = 0; 
		do{
			owner.encryptAndSendData(ts, streamId, ts%24, ts, ts);
			ts++;
			//Thread.sleep(2000);
		}
		while (true);	
	}
	
	public void sendDataFromFile(String fileName) throws Exception{
		ObjectInputStream ois = new ObjectInputStream(new FileInputStream(fileName));
		ArrayList<StockEvent> o = (ArrayList<StockEvent>)ois.readObject(); 		
		ois.close();		
		for (int i=200; i<300; i++){
			this.oos.writeObject(o.get(i));
			this.oos.flush();	
			Thread.sleep(100);
		}
	}
	
	public void sendDataFromFiles(String fileName, int rate) throws Exception{
		long start = System.currentTimeMillis();
		long end; 
		int count=0; 
		for (int i = 1; i <= 50; i++) {
			ObjectInputStream ois = new ObjectInputStream(new FileInputStream(
					fileName+"_"+i));
			ArrayList<StockEvent> o = (ArrayList<StockEvent>) ois.readObject();
			ois.close();			
			for (int j=0; j<o.size(); j++){
				this.oos.writeObject(o.get(j));
				this.oos.flush();
				count++; 				 
				if (count==rate){
					end = System.currentTimeMillis();
					Thread.sleep(1000-(end-start));
					count=0; 
					start = System.currentTimeMillis(); 
				}				
			}			
		}
	}
	public void sendDataFromFilePart(String fileName, int index, int rate) throws Exception{
		long start = System.currentTimeMillis();
		long end; 
		int count=0; 
		
			ObjectInputStream ois = new ObjectInputStream(new FileInputStream(
					fileName+"_"+index));
			ArrayList<StockEvent> o = (ArrayList<StockEvent>) ois.readObject();
			ois.close();
			for (int j=0; j<o.size(); j++){
				this.oos.writeObject(o.get(j));
				this.oos.flush();
				count++; 				 
				if (count==rate){
					end = System.currentTimeMillis();
					Thread.sleep(1000-(end-start));
					count=0; 
					start = System.currentTimeMillis(); 
				}				
			}			
		
	}
	public byte[] getPK(){
		return this.pK;
	}
	
	public byte[] getDetK(){
		return this.detK; 
	}
	
	public byte[] generateUserWindowKey(int streamId, int windowSize){
		return this.cryptoHandler.generateUserWindowKey_Aggregate1(streamId, windowSize);
	}
	
	public byte[] generateTransformKey(String policy, int streamId, int userId){
		return this.cryptoHandler.generateTransformKey(streamId, policy, userId);
	}
	
	public byte[] generateUserDecKey(int streamId, int userId){
		return this.cryptoHandler.generateUserDecryptionKey(streamId, userId);
	}
	
	public byte[] generateUserDetKey(int streamId){
		return this.cryptoHandler.generateUserDetKey(streamId);
	}
	
	public static void main(String[] args) throws Exception{
		int streamId = new Integer(args[0]).intValue();
		Properties prop = new Properties();
		int userMaxId=0; 
		try {
			prop.load(new FileInputStream("attributes.properties"));
			userMaxId = new Integer(prop.getProperty("maxUserId")).intValue();
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		Timer t = new Timer("Owner task", true);    	
    	t.scheduleAtFixedRate(new TimerTask(){
    		public void run(){
    			Stat.aggregateAndResetAll("ownerThroughput");     			
    		}
    	}, 0, 1000);
    	
		
		EPOwnerClient owner = new EPOwnerClient(streamId); 
				
		//simpleClient(streamId,0);
		//aggregateClient_2(owner,streamId,0,4);
		//aggregateClient_2(owner,streamId,1,3);
		//aggregateClient_3(owner,streamId,0,4,userMaxId);
		//aggregateClient_3(owner,streamId,1,2,userMaxId);
		//int userId = new Integer(args[1]).intValue();
		//aggregateClient_2(owner,streamId,0,4);		
		for (int i=50; i<100; i++){			
				simpleClient(owner, streamId, i);
				owner.waitForAck();			
		}
		//joinClient(owner,streamId,userId);
		//sendData(owner,streamId);
		owner.sendDataFromFile("enc_file_"+streamId);
		do{
			Thread.sleep(10000);
		}
		while (true);
	}
	
	//just like for simple client, 
	public static void joinClient(EPOwnerClient owner, int streamId, int userId){				    	
		owner.sendPolicy(initJoinClient(owner,streamId,userId));				
	}
	
	public static void aggregateClient_3(EPOwnerClient owner, int streamId,
			int userId, int windowSize, int maxUserId) {
		owner.sendPolicy(initAggregateClient_3(owner, streamId, userId,
				windowSize, maxUserId));
	}
	
	public static void aggregateClient_2(EPOwnerClient owner,int streamId, int userId, int windowSize){						
		owner.sendPolicy(initAggregateClient_2(owner,streamId, userId, windowSize));				
	}
	
	public static PolicyMessage initAggregateClient_1(EPOwnerClient owner,int streamId, int userId, int windowSize){		
		
		byte[] pK = owner.getPK(); 
		byte[] decK = owner.generateUserDecKey(streamId, userId); 
		byte[] detK = owner.getDetK();  				
		byte[] tK = owner.generateTransformKey("ws_" + windowSize
				+ " and stockId_" + streamId, streamId, userId); 			
		byte[] wK = owner.generateUserWindowKey(streamId, windowSize);
		
		byte[][] pKs = new byte[1][pK.length]; pKs[0] = pK;
		byte[][] decKs = new byte[1][decK.length]; decKs[0] = decK;
		byte[][] detKs = new byte[1][detK.length]; detKs[0] = detK;
		byte[][] tKs = new byte[1][tK.length]; tKs[0] = tK; 
		byte[][] wKs = new byte[1][wK.length]; wKs[0] = wK;
			
		int[] sids = {streamId}; int[] uids = {userId};  
		PolicyMessage selectPricePolicy = new PolicyMessage(sids, uids, pKs,
				tKs, detKs, decKs, wKs, windowSize, PolicyMessage.SELECT_ALL,
				PolicyMessage.AGG_QUERY_1,
				"select timestamp, realTime, stockId, price('ws=" + windowSize
						+ "'), volume('ws=" + windowSize
						+ "') from StockEvent(stockId=" + streamId
						+ ").win:length_batch(" + windowSize + ")");
		return 		selectPricePolicy;		
	}
	
public static PolicyMessage initAggregateClient_1(EPOwnerClient owner,int streamId, PolicyInfo pi){		
		
		byte[] pK = owner.getPK(); 
		byte[] decK = owner.generateUserDecKey(streamId, pi.getUserId()); 
		byte[] detK = owner.getDetK();  				
		byte[] tK = owner.generateTransformKey("ws_" + pi.getWindowSize()
				+ " and stockId_" + streamId, streamId, pi.getUserId()); 			
		byte[] wK = owner.generateUserWindowKey(streamId, pi.getWindowSize());
		
		byte[][] pKs = new byte[1][pK.length]; pKs[0] = pK;
		byte[][] decKs = new byte[1][decK.length]; decKs[0] = decK;
		byte[][] detKs = new byte[1][detK.length]; detKs[0] = detK;
		byte[][] tKs = new byte[1][tK.length]; tKs[0] = tK; 
		byte[][] wKs = new byte[1][wK.length]; wKs[0] = wK;
			
		int[] sids = {streamId}; int[] uids = {pi.getUserId()};  
		PolicyMessage selectPricePolicy = new PolicyMessage(sids, uids, pKs,
				tKs, detKs, decKs, wKs, pi.getWindowSize(), PolicyMessage.SELECT_ALL,
				PolicyMessage.AGG_QUERY_1,
				"select timestamp, realTime, stockId, price('ws=" + pi.getWindowSize()
						+ "'), volume('ws=" + pi.getWindowSize()
						+ "') from StockEvent(stockId=" + streamId
						+ ").win:length_batch(" + pi.getWindowSize() + ")");
		return 		selectPricePolicy;		
	}

	public static PolicyMessage initAggregateClient_2(EPOwnerClient owner,int streamId, int userId, int windowSize){		
		
		byte[] pK = owner.getPK(); 
		byte[] decK = owner.generateUserDecKey(streamId, userId); 
		byte[] detK = owner.getDetK();
		byte[] tK = owner.generateTransformKey("ws_" + windowSize
				+ " and stockId_" + streamId, streamId, userId); 					
		
		byte[][] pKs = new byte[1][pK.length]; pKs[0] = pK;
		byte[][] decKs = new byte[1][decK.length]; decKs[0] = decK;
		byte[][] detKs = new byte[1][detK.length]; detKs[0] = detK;
		byte[][] tKs = new byte[1][tK.length]; tKs[0] = tK; 
		
		int[] sids = {streamId}; int[] uids = {userId};  
		PolicyMessage selectPricePolicy = new PolicyMessage(sids, uids, pKs,
				tKs, detKs, decKs, null, windowSize, PolicyMessage.SELECT_ALL,
				PolicyMessage.AGG_QUERY_2,
				"select timestamp, realTime, stockId, price('ws=" + windowSize
						+ "'), volume('ws=" + windowSize
						+ "') from StockEvent(stockId=" + streamId
						+ ").win:length_batch(" + windowSize + ")");
		return 			selectPricePolicy;	
	}
	
	public static PolicyMessage initAggregateClient_2(EPOwnerClient owner,int streamId, PolicyInfo pi){		
		byte[] pK = owner.getPK(); 
		byte[] decK = owner.generateUserDecKey(streamId, pi.getUserId()); 
		byte[] detK = owner.getDetK();
		byte[] tK = owner.generateTransformKey(pi.getPolicy(), streamId, pi.getUserId()); 					
		
		byte[][] pKs = new byte[1][pK.length]; pKs[0] = pK;
		byte[][] decKs = new byte[1][decK.length]; decKs[0] = decK;
		byte[][] detKs = new byte[1][detK.length]; detKs[0] = detK;
		byte[][] tKs = new byte[1][tK.length]; tKs[0] = tK; 
		
		int[] sids = {streamId}; int[] uids = {pi.getUserId()};  
		PolicyMessage selectPricePolicy = new PolicyMessage(sids, uids, pKs,
				tKs, detKs, decKs, null, pi.getWindowSize(), PolicyMessage.SELECT_ALL,
				PolicyMessage.AGG_QUERY_2,
				"select timestamp, realTime, stockId, price('ws=" + pi.getWindowSize()
						+ "'), volume('ws=" + pi.getWindowSize()
						+ "') from StockEvent(" + pi.getSelectCondition()
						+ ").win:length_batch(" + pi.getWindowSize() + ")");
		
		return 			selectPricePolicy;	
	}
	
	public static PolicyMessage initAggregateClient_3(EPOwnerClient owner,int streamId, int userId, int windowSize, int maxUserId){		
		
		byte[] pK = owner.getPK(); 
		byte[] decK = owner.generateUserDecKey(streamId, userId); 
		byte[] decK1 = owner.generateUserDecKey(streamId, userId+maxUserId); 
		byte[] detK = owner.getDetK();
		byte[] tK = owner.generateTransformKey("ws_x and stockId_"+streamId, streamId, userId); 					
		byte[] tK1 = owner.generateTransformKey("ws_x and (timestamp % "
				+ windowSize + " = " + (windowSize - 1) + ")", streamId, userId
				+ maxUserId);
		
		byte[][] pKs = new byte[1][pK.length];
		pKs[0] = pK;
		byte[][] decKs = new byte[2][decK.length];
		decKs[0] = decK;
		decKs[1] = decK1;
		byte[][] detKs = new byte[1][detK.length];
		detKs[0] = detK;
		byte[][] tKs = new byte[2][tK.length];
		tKs[0] = tK;
		tKs[1] = tK1; 
		
		int[] sids = {streamId}; int[] uids = {userId};
		PolicyMessage selectPricePolicy = new PolicyMessage(sids, uids, pKs,
				tKs, detKs, decKs, null, windowSize, PolicyMessage.SELECT_ALL,
				PolicyMessage.AGG_QUERY_3,
				"select timestamp, realTime, stockId, price('ws=x'), "
						+ "volume('ws=x') from StockEvent(stockId=" + streamId
						+ ").win:length_batch(" + windowSize + ")");
		selectPricePolicy.setOtherUserId(userId+maxUserId);
		return selectPricePolicy;			
	}
	
	public static PolicyMessage initAggregateClient_3(EPOwnerClient owner,int streamId, PolicyInfo pi, int maxUserId){		
		
		int userId = pi.getUserId();
		int windowSize = pi.getWindowSize();
		byte[] pK = owner.getPK(); 
		byte[] decK = owner.generateUserDecKey(streamId, userId); 
		byte[] decK1 = owner.generateUserDecKey(streamId, userId+maxUserId); 
		byte[] detK = owner.getDetK();
		byte[] tK = owner.generateTransformKey("ws_x and stockId_"+streamId, streamId, userId); 					
		byte[] tK1 = owner.generateTransformKey("ws_x and (timestamp % "
				+ windowSize + " = " + (windowSize - 1) + ")", streamId, userId
				+ maxUserId);
		
		byte[][] pKs = new byte[1][pK.length];
		pKs[0] = pK;
		byte[][] decKs = new byte[2][decK.length];
		decKs[0] = decK;
		decKs[1] = decK1;
		byte[][] detKs = new byte[1][detK.length];
		detKs[0] = detK;
		byte[][] tKs = new byte[2][tK.length];
		tKs[0] = tK;
		tKs[1] = tK1; 
		
		int[] sids = {streamId}; int[] uids = {userId};
		PolicyMessage selectPricePolicy = new PolicyMessage(sids, uids, pKs,
				tKs, detKs, decKs, null, windowSize, PolicyMessage.SELECT_ALL,
				PolicyMessage.AGG_QUERY_3,
				"select timestamp, realTime, stockId, price('ws=x'), "
						+ "volume('ws=x') from StockEvent(stockId=" + streamId
						+ ").win:length_batch(" + windowSize + ")");
		selectPricePolicy.setOtherUserId(userId+maxUserId);
		return selectPricePolicy;			
	}
	
	public static PolicyMessage initJoinClient(EPOwnerClient owner, int streamId, int userId){		
		
    	byte[] pK = owner.getPK(); 
		byte[] decK = owner.generateUserDecKey(streamId, userId); 
		byte[] detK = owner.getDetK();
		byte[] tK = owner.generateTransformKey("stockId_"+streamId, streamId, userId); 	
		byte[][] pKs = new byte[1][pK.length]; pKs[0] = pK;
		byte[][] decKs = new byte[1][decK.length]; decKs[0] = decK;
		byte[][] detKs = new byte[1][detK.length]; detKs[0] = detK;
		byte[][] tKs = new byte[1][tK.length]; tKs[0] = tK; 
		
		int[] sids = {streamId}; int[] uids = {userId};  
		PolicyMessage selectPricePolicy = new PolicyMessage(sids, uids, pKs,
				tKs, detKs, decKs, null, 0, PolicyMessage.SELECT_VOLUME,
				PolicyMessage.JOIN_QUERY,"");
		return selectPricePolicy;			
	}
	
	public static PolicyMessage initJoinClient(EPOwnerClient owner, int streamId, PolicyInfo pi){		
    	byte[] pK = owner.getPK(); 
		byte[] decK = owner.generateUserDecKey(streamId, pi.getUserId()); 
		byte[] detK = owner.getDetK();
		byte[] tK = owner.generateTransformKey(pi.getPolicy(), streamId, pi.getUserId()); 	
		byte[][] pKs = new byte[1][pK.length]; pKs[0] = pK;
		byte[][] decKs = new byte[1][decK.length]; decKs[0] = decK;
		byte[][] detKs = new byte[1][detK.length]; detKs[0] = detK;
		byte[][] tKs = new byte[1][tK.length]; tKs[0] = tK; 
		
		int[] sids = {streamId}; int[] uids = {pi.getUserId()};  
		PolicyMessage selectPricePolicy = new PolicyMessage(sids, uids, pKs,
				tKs, detKs, decKs, null, 0, PolicyMessage.SELECT_VOLUME,
				PolicyMessage.JOIN_QUERY,pi.getSelectCondition());
		
		return selectPricePolicy;			
	}
	public static void aggregateClient_1(EPOwnerClient owner,int streamId, int userId, int windowSize){		
					
		owner.sendPolicy(initAggregateClient_1(owner,streamId,userId, windowSize));				
	}
	
	
	public static PolicyMessage initSimpleClient(EPOwnerClient owner,int streamId, int userId){
		byte[] pK = owner.getPK(); 
		byte[] decK = owner.generateUserDecKey(streamId, userId); 
		byte[] detK = owner.getDetK();
		byte[] tK = owner.generateTransformKey("stockId_"+streamId, streamId, userId); 	
		byte[][] pKs = new byte[1][pK.length]; pKs[0] = pK;
		byte[][] decKs = new byte[1][decK.length]; decKs[0] = decK;
		byte[][] detKs = new byte[1][detK.length]; detKs[0] = detK;
		byte[][] tKs = new byte[1][tK.length]; tKs[0] = tK; 
		
		int[] sids = {streamId}; int[] uids = {userId};  
		PolicyMessage selectPricePolicy = new PolicyMessage(sids, uids, pKs,
				tKs, detKs, decKs, null, 0, PolicyMessage.SELECT_ALL,
				PolicyMessage.SIMPLE_QUERY,
				"select timestamp, realTime, stockId, price('ws=1'), "
						+ "volume('ws=1') from StockEvent(stockId=" + streamId
						+ ")");
		
		return selectPricePolicy; 
	}
	
	public static PolicyMessage initSimpleClient(EPOwnerClient owner,int streamId, PolicyInfo pi){
		byte[] pK = owner.getPK(); 
		byte[] decK = owner.generateUserDecKey(streamId, pi.getUserId()); 
		byte[] detK = owner.getDetK();
		byte[] tK = owner.generateTransformKey(pi.getPolicy(), streamId, pi.getUserId()); 	
		byte[][] pKs = new byte[1][pK.length]; pKs[0] = pK;
		byte[][] decKs = new byte[1][decK.length]; decKs[0] = decK;
		byte[][] detKs = new byte[1][detK.length]; detKs[0] = detK;
		byte[][] tKs = new byte[1][tK.length]; tKs[0] = tK; 
		
		int[] sids = {streamId}; int[] uids = {pi.getUserId()};  
		PolicyMessage selectPricePolicy = new PolicyMessage(sids, uids, pKs,
				tKs, detKs, decKs, null, 0, PolicyMessage.SELECT_ALL,
				PolicyMessage.SIMPLE_QUERY,
				"select timestamp, realTime, stockId, price('ws=1'), "
						+ "volume('ws=1') from StockEvent("+pi.getSelectCondition()+")");
		return selectPricePolicy; 
	}
	
	public static void simpleClient(EPOwnerClient owner,int streamId, int userId) throws Exception{
		//Read from file
		ObjectInputStream ois = new ObjectInputStream(new FileInputStream(
				"policy_file_" + streamId)); 
		ArrayList<PolicyMessage> policies = (ArrayList<PolicyMessage>)ois.readObject();
		owner.sendPolicy(policies.get(userId)); 
		log.info("SENT POLICY , user id = "+policies.get(userId).getUserIds()[0]+" query = "+policies.get(userId).getQueryExpression()); 
		//owner.sendPolicy(initSimpleClient(owner,streamId,userId));				
	}
			
}
